package io.kiki.sba.registry.server.metadata.multi.cluster.remote;

import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.GenericResponse;
import io.kiki.sba.registry.common.model.Node.NodeType;
import io.kiki.sba.registry.common.model.multi.cluster.DataCenter;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.server.metadata.bootstrap.ExecutorManager;
import io.kiki.sba.registry.server.metadata.bootstrap.config.ServerConfig;
import io.kiki.sba.registry.server.metadata.metadataServer.CurrentNodeClusterMetadataServer;
import io.kiki.sba.registry.server.shared.remoting.AbstractServerHandler;
import io.kiki.sba.registry.server.shared.slot.SlotsUtils;
import io.kiki.sba.registry.store.api.config.DefaultCommonConfig;
import io.kiki.sba.registry.store.api.elector.Leader;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.Executor;

@Setter
@Getter
public class RemoteClusterSlotSyncHandler extends AbstractServerHandler<RemoteClusterSlotSyncRequest> {

    private static final Logger logger = LoggerFactory.getLogger(RemoteClusterSlotSyncHandler.class);
    @Autowired
    private ExecutorManager executorManager;

    @Autowired
    private CurrentNodeClusterMetadataServer currentDcMetaServer;

    @Autowired
    private MetadataLeaderService metadataLeaderService;

    @Autowired
    private ServerConfig serverConfig;

    @Autowired
    private DefaultCommonConfig defaultCommonConfig;

    @Override
    public Class interest() {
        return RemoteClusterSlotSyncRequest.class;
    }

    @Override
    protected NodeType getConnectNodeType() {
        return NodeType.central_control;
    }


    @Override
    public GenericResponse<RemoteClusterSlotSyncResponse> doHandle(Channel channel, RemoteClusterSlotSyncRequest request) {

        String clusterId = defaultCommonConfig.getDefaultClusterId();
        Leader leader = metadataLeaderService.getLeaderInfo();
        // wrong leader
        if (!metadataLeaderService.amILeader()) {
            logger.error("request: {} sync on follower, leader is:{}", request, leader.getLeader());
            return new GenericResponse<RemoteClusterSlotSyncResponse>().fillFailData(RemoteClusterSlotSyncResponse.wrongLeader(leader.getLeader(), leader.getEpoch()));
        }

        // leader not warmuped
        if (!metadataLeaderService.amIStableAsLeader()) {
            logger.error("request: {} sync on leader not warmuped.", request);
            return new GenericResponse<RemoteClusterSlotSyncResponse>().fillFailData(RemoteClusterSlotSyncResponse.leaderNotWarmuped(leader.getLeader(), leader.getEpoch()));
        }

        Slots slots = currentDcMetaServer.getSlots();
        if (!SlotsUtils.isValidSlots(slots)) {
            return new GenericResponse<RemoteClusterSlotSyncResponse>().fillFailed("slot-table not valid, check meta-server log for detail");
        }

        // check slot table epoch
        if (request.getSlotsEpoch() > slots.getEpoch()) {
            // it should not happen, print error log and return false, restart meta leader;
            logger.error("[conflict]request: {} slotEpoch > local.slotEpoch: {}", request, slots.getEpoch());
            return new GenericResponse<RemoteClusterSlotSyncResponse>().fillFailed("slotEpoch conflict");
        } else
            if (request.getSlotsEpoch() == slots.getEpoch()) {
                return new GenericResponse<RemoteClusterSlotSyncResponse>().fillSucceed(RemoteClusterSlotSyncResponse.notUpgrade(leader.getLeader(), leader.getEpoch(), new DataCenter(clusterId, serverConfig.getLocalDataCenterZones())));
            } else {
                Slots resp = slots.filterLeaderInfo();
                logger.info("sync request epoch:{} < newslotsEpoch:{}, leader:{}, leaderEpoch:{}, Slots upgrade: {}", request, slots.getEpoch(), leader.getLeader(), leader.getEpoch(), resp);
                return new GenericResponse<RemoteClusterSlotSyncResponse>().fillSucceed(RemoteClusterSlotSyncResponse.upgrade(leader.getLeader(), leader.getEpoch(), resp, new DataCenter(clusterId, serverConfig.getLocalDataCenterZones())));
            }
    }

    @Override
    public Executor getExecutor() {
        return executorManager.getRemoteClusterHandlerExecutor();
    }
}
